package com.example.tcp.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.ReadTimeoutHandler;
import io.netty.util.concurrent.Promise;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.LinkedList;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;

/**
 * 无限重连client
 */
@Slf4j
@ChannelHandler.Sharable
public class RetryTcpClient extends ChannelInboundHandlerAdapter {

    private static final EventCallback NOOP = new EventCallback() {};

    private enum CLIENT_STATUS { INIT, CONNECTED, RECONNECT, CLOSED }

    private Channel channel;

    private Bootstrap bootstrap = new Bootstrap();

    /**
     * 事件回调
     */
    private EventCallback eventCallback;

//    private IDequeStore<Message> iDequeStore;

    private AtomicInteger retryCount = new AtomicInteger(0);

    private RetryStrategy retryStrategy;

    private final EventLoopGroup workerGroup = new NioEventLoopGroup();
    private final ScheduledExecutorService scheduledExecutorService = Executors.newSingleThreadScheduledExecutor();

    private AtomicReference<CLIENT_STATUS> status = new AtomicReference<>(CLIENT_STATUS.INIT);

    public static class Builder {
        private int connectTimeoutMs = 60000;
        private long readTimeoutMs = 0;
        private SocketAddress addr;
        private Consumer<SocketChannel> onChannelInitializer;
        private EventCallback eventCallback = NOOP;
//        private IDequeStore<Message> iDequeStore;
        private RetryStrategy retryStrategy;

        public Builder address(String host, int port) {
            this.addr = new InetSocketAddress( host, port );
            return this;
        }

        public Builder connectTimeoutMs(int connectTimeoutMs) {
            this.connectTimeoutMs = connectTimeoutMs;
            return this;
        }

        public Builder readTimeoutMs(long readTimeoutMs) {
            this.readTimeoutMs = readTimeoutMs;
            return this;
        }

        public Builder onChannelInitializer(Consumer<SocketChannel> onChannelInitializer) {
            this.onChannelInitializer = onChannelInitializer;
            return this;
        }

        public Builder eventCallback(EventCallback eventCallback) {
            this.eventCallback = eventCallback;
            return this;
        }

//        public Builder store(IDequeStore<Message> iDequeStore) {
//            this.iDequeStore = iDequeStore;
//            return this;
//        }

        public Builder retryStrategy(RetryStrategy retryStrategy) {
            this.retryStrategy = retryStrategy;
            return this;
        }

        public RetryTcpClient build() {

            if(connectTimeoutMs < 0) {
                throw new IllegalArgumentException("connectTimeoutMs must great than 0");
            }

            if(readTimeoutMs < 0) {
                throw new IllegalArgumentException("readTimeoutMs must great than 0");
            }

            if(eventCallback == null) {
                eventCallback = NOOP;
            }

//            if(iDequeStore == null) {
//                iDequeStore = new MemoryDequeStore<>();
//            }

            if(retryStrategy == null) {
                retryStrategy = RetryStrategy.DEFAULT;
            }

            return new RetryTcpClient(addr, onChannelInitializer,  retryStrategy, connectTimeoutMs, readTimeoutMs, eventCallback);
        }
    }

    private RetryTcpClient(SocketAddress addr, Consumer<SocketChannel> onChannelInitializer, RetryStrategy retryStrategy, int connectTimeoutMs, long readTimeoutMs, EventCallback eventCallback) {
        this.eventCallback = eventCallback;
//        this.iDequeStore = iDequeStore;
        this.retryStrategy = retryStrategy;
        bootstrap.group(workerGroup);
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeoutMs);
        bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
        bootstrap.option(ChannelOption.TCP_NODELAY, true);
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            public void initChannel(SocketChannel ch) throws Exception {
                if(onChannelInitializer != null) {
                    onChannelInitializer.accept(ch);
                }
                if(readTimeoutMs > 0) {
                    ch.pipeline().addLast("Timeout", new ReadTimeoutHandler(readTimeoutMs, TimeUnit.MILLISECONDS));
                }
                ch.pipeline().addLast(RetryTcpClient.this);
            }
        });
        doConnect(addr);
    }

    private void doConnect(SocketAddress addr) {
        if(status.get() != CLIENT_STATUS.CLOSED) {
            try {
                ChannelFuture f = bootstrap.connect(addr);
                f.addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        if( !future.isSuccess() ) {
                            eventCallback.onRetryCount(RetryTcpClient.this, retryCount.get());
                            future.channel().close();
                            scheduleConnect(addr, future.cause());
                        } else {
                            retryCount.set(0);
                            channel = future.channel();
                            if(status.compareAndSet(CLIENT_STATUS.INIT, CLIENT_STATUS.CONNECTED)) {
                                eventCallback.onConnected(RetryTcpClient.this);
//                                new Thread(() -> writeMessage()).start();
                            }
                            channel.closeFuture().addListener(f -> {
                                scheduleConnect(addr, f.cause());
                            });
                        }
                    }
                });
            } catch (Throwable e) {
                scheduleConnect(addr, e);
            }
        }
    }

    private void scheduleConnect(SocketAddress addr, Throwable e) {
        if(status.get() != CLIENT_STATUS.CLOSED && canRetry(e)) {
            if(status.compareAndSet(CLIENT_STATUS.CONNECTED, CLIENT_STATUS.RECONNECT)) {
                eventCallback.onStartReconnect(this);
            }
            scheduledExecutorService.schedule(() -> doConnect(addr), retryStrategy.backoffStrategy.delayBeforeNextRetry(retryCount.get()), TimeUnit.MILLISECONDS);
        } else {
            close();
        }
    }

    private boolean canRetry(Throwable e) {
        return retryCount.incrementAndGet() < retryStrategy.maxRetry && retryStrategy.retryCondition.shouldRetry(e, retryCount.get());
    }

    public void close() {
        if(status.getAndUpdate( s -> CLIENT_STATUS.CLOSED) != CLIENT_STATUS.CLOSED) {
//            iDequeStore.close();
            if(channel != null) {
                channel.close();
            }
            workerGroup.shutdownGracefully();
            scheduledExecutorService.shutdown();
        }
    }

    public ChannelFuture send(Message msg) {
//        return iDequeStore.enqueue(msg);
        if(channel != null && channel.isWritable()) {
            return channel.writeAndFlush(Unpooled.wrappedBuffer(msg.serialize()));
        }

        return null;
    }

    public boolean isWritable() {
        return channel != null && channel.isWritable();
    }

    public boolean isClosed() {
        return status.get() == CLIENT_STATUS.CLOSED;
    }

//    private void writeMessage() {
//        while (status.get() != CLIENT_STATUS.CLOSED) {
//            try {
//                if(isWritable()) {
//                    Message msg = iDequeStore.dequeue();
//
//                    if(msg == null) {
//                        continue;
//                    }
//
//                    channel.writeAndFlush(Unpooled.wrappedBuffer(msg.serialize())).addListener(future -> {
//                        if(future.isSuccess()) {
//                            eventCallback.onWriteMessageComplete(this, msg);
//                        } else {
//                            eventCallback.onWriteMessageError(this, msg);
//                        }
//                    });
//                } else {
//                    Thread.sleep(500);
//                }
//            } catch (Throwable e) {
//                log.error("catch error", e);
//            }
//        }
//    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        eventCallback.onReadMessage(this, ctx, msg);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
         if(status.compareAndSet(CLIENT_STATUS.RECONNECT, CLIENT_STATUS.CONNECTED)) {
            eventCallback.onReconnectSuccess(this);
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        if(status.get() == CLIENT_STATUS.CLOSED) {
            eventCallback.onClosed(this);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        eventCallback.onExceptionCaught(ctx, cause);
        ctx.close();
    }

    public interface Message {
        byte[] serialize();
        Message deserialize(byte[] value);
    }

    public static class ByteBufMessage implements RetryTcpClient.Message {

        private ByteBuf buf;

        public ByteBufMessage() {
            this.buf = Unpooled.EMPTY_BUFFER;
        }

        public ByteBufMessage(ByteBuf buf) {
            this.buf = buf;
        }

        public ByteBuf getBuf() {
            return buf;
        }

        @Override
        public byte[] serialize() {
            return ByteBufUtil.getBytes(buf);
        }

        @Override
        public ByteBufMessage deserialize(byte[] value) {
            this.buf = Unpooled.copiedBuffer(value);
            return this;
        }
    }


    public interface EventCallback {

        /**
         * 第一次连上服务器回调
         * @param client
         */
        default void onConnected(RetryTcpClient client) {
            log.info("onConnected");
        }

        /**
         * 最后断开服务器连接回调
         * @param client
         */
        default void onClosed(RetryTcpClient client) {
            log.info("onClosed");
        }

        /**
         * 开始重连回调
         * @param client
         */
        default void onStartReconnect(RetryTcpClient client) {
            log.info("onStartReconnect");
        }


        default void onRetryCount(RetryTcpClient client, int count) {
            log.info("try: {}", count);
        }

        /**
         * 重连结束回调，
         * @param client
         */
        default void onReconnectSuccess(RetryTcpClient client) {
            log.info("onReconnectSuccess");
        }

        /**
         * 收到消息回调
         * @param client
         * @param ctx
         * @param msg
         */
        default void onReadMessage(RetryTcpClient client, ChannelHandlerContext ctx, Object msg) {}

        /**
         * 写链路成功回调
         * @param client
         * @param msg
         */
        default void onWriteMessageComplete(RetryTcpClient client, Message msg) {

        }

        default void onWriteMessageError(RetryTcpClient client, Message msg) {
            log.error("onWriteMessageError");
        }

        default void onExceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            log.error("onExceptionCaught", cause);
        }
    }

    public static class RetryStrategy {

        public static final RetryStrategy DEFAULT = new RetryStrategy(RetryCondition.ALWAYS_RETRY_CONDITION, BackoffStrategy.DELAY_LINEAR, Integer.MAX_VALUE);

        private RetryCondition retryCondition;
        private BackoffStrategy backoffStrategy;
        private int maxRetry;

        public RetryStrategy(RetryCondition retryCondition, BackoffStrategy backoffStrategy, int maxRetry) {

            if(retryCondition == null) {
                retryCondition = RetryCondition.ALWAYS_RETRY_CONDITION;
            }

            if(backoffStrategy == null) {
                backoffStrategy = BackoffStrategy.NO_DELAY;
            }
            if (maxRetry < 0) {
                throw new IllegalArgumentException( "Please provide a non-negative value for maxRetry.");
            }
            this.retryCondition = retryCondition;
            this.backoffStrategy = backoffStrategy;
            this.maxRetry = maxRetry;
        }

        public interface RetryCondition {

            public static final RetryCondition ALWAYS_RETRY_CONDITION = new RetryCondition() {
                @Override
                public boolean shouldRetry(Throwable e, int retryCount) {
                    return true;
                }
            };

            boolean shouldRetry(Throwable e, int retryCount);
        }

        public interface BackoffStrategy {

            public static final BackoffStrategy NO_DELAY = new BackoffStrategy() {
                @Override
                public long delayBeforeNextRetry(int retryCount) {
                    return 0;
                }
            };

            /**
             * 线性增加,每尝试一次增加一秒延迟如[1s,2s,3s.....]
             */
            public static final BackoffStrategy DELAY_LINEAR = new BackoffStrategy() {
                @Override
                public long delayBeforeNextRetry(int retryCount) {
                    return retryCount * 1000L;
                }
            };

            long delayBeforeNextRetry(int retryCount);
        }
    }




}
